package com.gome.ss.common;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;

import com.gome.ss.core.download.SpiderDataMap;

public class FlowCrawlHbaseHandle {
    public static Configuration configuration;  
    static {
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        configuration.set("hbase.zookeeper.quorum", "10.58.222.102");
    }
    
    public static void createTable(String tableName) {
        try {
            HBaseAdmin hBaseAdmin = new HBaseAdmin(configuration);
            if (hBaseAdmin.tableExists(tableName)) {
                hBaseAdmin.disableTable(tableName);
                hBaseAdmin.deleteTable(tableName);
            }
            String coprocessClassName = "org.apache.hadoop.hbase.coprocessor.AggregateImplementation";
            HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
            tableDescriptor.addFamily(new HColumnDescriptor("info"));
            tableDescriptor.addFamily(new HColumnDescriptor("parent"));
            tableDescriptor.addCoprocessor(coprocessClassName);
            tableDescriptor.setCompactionEnabled(true);
            hBaseAdmin.createTable(tableDescriptor);
        } catch (MasterNotRunningException e) {
            e.printStackTrace();
        } catch (ZooKeeperConnectionException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    public static void dropTable(String tableName) {
        try {
            HBaseAdmin admin = new HBaseAdmin(configuration);
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        } catch (MasterNotRunningException e) {
            e.printStackTrace();
        } catch (ZooKeeperConnectionException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    public static void insertData(String tableName, SpiderDataMap dataMap) {
		try {
			HTable table = new HTable(configuration, tableName);
	        Put put = new Put(((String)dataMap.get("id")).getBytes());
	        
	        put.add("info".getBytes(), "id".getBytes(), ((String)dataMap.get("id")).getBytes());
	        put.add("info".getBytes(), "stepID".getBytes(), (String.valueOf((Integer)dataMap.get("stepID"))).getBytes());
	        put.add("info".getBytes(), "title".getBytes(), ((String)dataMap.get("title")).getBytes());
	        put.add("info".getBytes(), "url".getBytes(), ((String)dataMap.get("url")).getBytes());
	        put.add("info".getBytes(), "cookies".getBytes(), ((String)dataMap.get("cookies")).getBytes());
	        put.add("info".getBytes(), "pageSource".getBytes(), ((String)dataMap.get("pageSource")).getBytes());
	        put.add("info".getBytes(), "failedWait".getBytes(), ((String)dataMap.get("failedWait")).getBytes());
	        put.add("info".getBytes(), "failedAction".getBytes(), ((String)dataMap.get("failedAction")).getBytes());
	        put.add("info".getBytes(), "crawlName".getBytes(), ((String)dataMap.get("crawlName")).getBytes());
	        
	        Map<String, String> parentInfos = (Map<String, String>) dataMap.get("parent");
	        if (parentInfos != null) {
		        put.add("parent".getBytes(), "id".getBytes(), parentInfos.get("id").getBytes());
		        put.add("parent".getBytes(), "url".getBytes(), parentInfos.get("url").getBytes());
		        put.add("parent".getBytes(), "qn".getBytes(), parentInfos.get("qn").getBytes());
		        put.add("parent".getBytes(), "stepID".getBytes(), parentInfos.get("stepID").getBytes());
		        put.add("parent".getBytes(), "stepName".getBytes(), parentInfos.get("stepName").getBytes());
		        put.add("parent".getBytes(), "cookies".getBytes(), parentInfos.get("cookies").getBytes());
	        }

	        table.put(put);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    }
}
